package com.guchenbo.example.kafka;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.io.resource.ClassPathResource;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

public class KafkaProducerTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        producer();
    }

    private static void producer() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");

        //        properties.put("acks", "1");
        //        properties.put("retries", 2);
        //        properties.put("enable.idempotence", "true");
        //        properties.put("retry.backoff.ms", 300);
        properties.put("max.request.size", 4194304);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        AtomicInteger cnt = new AtomicInteger();
        while (true) {
            Future<RecordMetadata> future = null;
            try {
                String v = "log gateway4 " + cnt.incrementAndGet();
                // 指定分区编号
                //                ProducerRecord record = new ProducerRecord("part4", 0, "testkey", v);
                // 指定分区key
                //                ProducerRecord record = new ProducerRecord("part4", "testkey", v);
                // 让kafka自己决定
                ProducerRecord record = new ProducerRecord("part4", v);
                future = kafkaProducer.send(record);

                RecordMetadata metadata = future.get();
                System.out.println("success send for " + cnt.get() + ", " + metadata);
            } catch (Exception e) {
                e.printStackTrace();
            }

            Thread.sleep(300);
        }
        //    for (int i = 1; i <= 6; i++) {
        //    }
        //    kafkaProducer.close();
    }

    public static String bigStr() {
        ClassPathResource resource = new ClassPathResource("/pic.png");
        System.out.println(resource.getAbsolutePath());
        return Base64.encode(resource.getFile());
    }

}
